package com.pokermind.dse.tcp.client;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.pokermind.dse.tcp.protocol.ProtocolMessage;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.CharsetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

/**
 * 简化版TCP客户端
 * 专门用于主程序包，不依赖测试框架
 */
public class SimpleTcpClient {
    
    private static final Logger logger = LoggerFactory.getLogger(SimpleTcpClient.class);
    private final ObjectMapper objectMapper = new ObjectMapper();
    
    private final String host;
    private final int port;
    private final String clientConnectionId;  // 客户端连接标识符
    
    private EventLoopGroup group;
    private Channel channel;
    private volatile boolean connected = false;
    
    // 存储待处理的请求 - 使用UUID字符串作为键
    private final ConcurrentHashMap<String, CompletableFuture<String>> pendingRequests = new ConcurrentHashMap<>();
    
    // 心跳保活相关
    private final ScheduledExecutorService heartbeatExecutor = Executors.newSingleThreadScheduledExecutor();
    private ScheduledFuture<?> heartbeatTask;
    private final long heartbeatInterval = 20; // 20秒发送一次心跳，小于服务器30秒超时
    
    public SimpleTcpClient(String host, int port) {
        this.host = host;
        this.port = port;
        this.clientConnectionId = "CONN_" + System.currentTimeMillis();  // 生成客户端连接标识符
    }
    
    public SimpleTcpClient(String host, int port, String connectionId) {
        this.host = host;
        this.port = port;
        this.clientConnectionId = connectionId != null ? connectionId : ("CONN_" + System.currentTimeMillis());
    }
    
    /**
     * 获取客户端连接标识符
     */
    public String getClientConnectionId() {
        return clientConnectionId;
    }
    
    public CompletableFuture<Void> connect() {
        CompletableFuture<Void> future = new CompletableFuture<>();
        
        group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap()
                .group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel ch) {
                        ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                            // 用于累积分包数据
                            private ByteBuf accumulatedBuffer;
                            
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                if (msg instanceof ByteBuf) {
                                    ByteBuf receivedBuf = (ByteBuf) msg;
                                    
                                    try {
                                        // 如果有累积缓冲区，先合并数据
                                        if (accumulatedBuffer != null) {
                                            // 创建新的缓冲区来合并数据
                                            ByteBuf combined = ctx.alloc().buffer(accumulatedBuffer.readableBytes() + receivedBuf.readableBytes());
                                            combined.writeBytes(accumulatedBuffer);
                                            combined.writeBytes(receivedBuf);
                                            
                                            // 释放旧的缓冲区
                                            accumulatedBuffer.release();
                                            accumulatedBuffer = combined;
                                        } else {
                                            // 保留一份数据的副本
                                            accumulatedBuffer = receivedBuf.retainedDuplicate();
                                        }
                                        
                                        // 尝试处理完整消息
                                        handleCompleteMessages(ctx);
                                        
                                    } finally {
                                        // 释放接收到的原始缓冲区
                                        receivedBuf.release();
                                    }
                                } else {
                                    ctx.fireChannelRead(msg);
                                }
                            }
                            
                            private void handleCompleteMessages(ChannelHandlerContext ctx) {
                                while (accumulatedBuffer != null && accumulatedBuffer.readableBytes() >= 6) {
                                    // 标记读取位置
                                    int readerIndex = accumulatedBuffer.readerIndex();
                                    
                                    // 读取协议头
                                    short protocolId = accumulatedBuffer.readShort();
                                    int bodyLength = accumulatedBuffer.readInt();
                                    
                                    // 检查消息体是否完整
                                    if (accumulatedBuffer.readableBytes() >= bodyLength) {
                                        // 重置读取位置到消息开始处，为handleResponse准备完整消息
                                        accumulatedBuffer.readerIndex(readerIndex);
                                        
                                        // 创建包含完整消息（协议头+消息体）的切片
                                        ByteBuf completeMessageSlice = accumulatedBuffer.readSlice(6 + bodyLength);
                                        
                                        // 处理完整消息
                                        try {
                                            handleResponse(completeMessageSlice);
                                        } catch (Exception e) {
                                            logger.error("Error handling response", e);
                                        }
                                        
                                        // 如果还有剩余数据，保留它
                                        if (accumulatedBuffer.isReadable()) {
                                            ByteBuf remaining = accumulatedBuffer.retainedSlice();
                                            accumulatedBuffer.release();
                                            accumulatedBuffer = remaining;
                                        } else {
                                            accumulatedBuffer.release();
                                            accumulatedBuffer = null;
                                        }
                                    } else {
                                        // 消息不完整，重置读取位置等待更多数据
                                        accumulatedBuffer.readerIndex(readerIndex);
                                        break;
                                    }
                                }
                            }
                            
                            @Override
                            public void channelActive(ChannelHandlerContext ctx) {
                                connected = true;
                                logger.info("Connected to TCP server: {}:{}", host, port);
                                startHeartbeat(); // 启动心跳保活
                                future.complete(null);
                            }
                            
                            @Override
                            public void channelInactive(ChannelHandlerContext ctx) {
                                connected = false;
                                stopHeartbeat(); // 停止心跳
                                logger.info("Connection to server closed");
                                
                                // 清理累积缓冲区
                                if (accumulatedBuffer != null) {
                                    accumulatedBuffer.release();
                                    accumulatedBuffer = null;
                                }
                            }
                            
                            @Override
                            public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
                                logger.error("Client error: {}", cause.getMessage());
                                future.completeExceptionally(cause);
                            }
                        });
                    }
                });
        
        bootstrap.connect(host, port).addListener((ChannelFutureListener) channelFuture -> {
            if (channelFuture.isSuccess()) {
                channel = channelFuture.channel();
            } else {
                future.completeExceptionally(channelFuture.cause());
            }
        });
        
        return future;
    }
    
    public boolean isConnected() {
        return connected && channel != null && channel.isActive();
    }
    
    public CompletableFuture<String> sendMessage(short protocolId, Object requestData, 
                                               long timeout, TimeUnit timeUnit) {
        if (!connected || channel == null || !channel.isActive()) {
            return CompletableFuture.failedFuture(new IllegalStateException("Client not connected"));
        }
        
        try {
            // 生成客户端请求ID - 使用UUID小写格式
            String clientRequestId = java.util.UUID.randomUUID().toString().replace("-", "");
            
            // 构造包含rqid和connId的请求包装对象
            Map<String, Object> requestWrapper = new LinkedHashMap<>();
            requestWrapper.put("rqid", clientRequestId);             // 关键：包含请求ID（UUID格式）
            requestWrapper.put("connId", getClientConnectionId());   // 客户端连接标识符
            requestWrapper.put("data", requestData);                // 实际业务数据
            
            // 序列化包装后的请求
            String jsonRequest = objectMapper.writeValueAsString(requestWrapper);
            byte[] jsonBytes = jsonRequest.getBytes(CharsetUtil.UTF_8);
            
            // 创建响应Future，使用clientRequestId作为键
            CompletableFuture<String> responseFuture = new CompletableFuture<>();
            pendingRequests.put(clientRequestId, responseFuture);   // 使用UUID格式的clientRequestId作为key
            
            // 设置超时处理
            responseFuture.orTimeout(timeUnit.toMillis(timeout), TimeUnit.MILLISECONDS)
                .whenComplete((result, throwable) -> {
                    pendingRequests.remove(clientRequestId);
                    if (throwable != null) {
                        logger.warn("Request {} timed out or failed: {}", clientRequestId, throwable.getMessage());
                    }
                });
            
            // 构造协议消息 (协议ID + 长度 + 包装后的JSON)
            ByteBuf buffer = Unpooled.buffer();
            buffer.writeShort(protocolId);           // 2字节协议ID
            buffer.writeInt(jsonBytes.length);       // 4字节长度
            buffer.writeBytes(jsonBytes);            // 包装后的JSON内容
            
            // 发送消息
            channel.writeAndFlush(buffer);
            
            if (logger.isDebugEnabled()) {
                logger.debug("Sent request: protocolId={}, requestId={}, dataSize={}", 
                    protocolId, clientRequestId, jsonBytes.length);
            }
            
            return responseFuture;
            
        } catch (Exception e) {
            return CompletableFuture.failedFuture(e);
        }
    }
    
    private void handleResponse(ByteBuf msg) {
        try {
            // 新的处理器保证传入的是完整消息，直接解析
            short protocolId = msg.readShort();
            int bodyLength = msg.readInt();
            
            byte[] bodyBytes = new byte[bodyLength];
            msg.readBytes(bodyBytes);
            String jsonBody = new String(bodyBytes, CharsetUtil.UTF_8);
            
            if (logger.isDebugEnabled()) {
                logger.debug("Received response: protocolId={}, bodyLength={}, body={}",
                    protocolId, bodyLength, jsonBody.length() > 200 ? jsonBody.substring(0, 200) + "..." : jsonBody);
            }
            
            // 精确匹配：根据响应中的rqid找到对应的Future
            try {
                JsonNode responseNode = objectMapper.readTree(jsonBody);
                if (responseNode.has("rqid")) {
                    String requestId = responseNode.get("rqid").asText();
                    
                    if (logger.isDebugEnabled()) {
                        logger.debug("Processing response: rqid={}, protocolId={}, pendingRequests.size={}", 
                            requestId, protocolId, pendingRequests.size());
                    }
                    
                    // 根据rqid精确匹配对应的Future
                    CompletableFuture<String> future = pendingRequests.remove(requestId);
                    if (future != null) {
                        future.complete(jsonBody);
                        if (logger.isDebugEnabled()) {
                            logger.debug("Response matched: rqid={}, protocolId={}", requestId, protocolId);
                        }
                    } else {
                        logger.error("No pending request found for rqid: {} (protocolId: {})", 
                            requestId, protocolId);
                    }
                } else {
                    logger.error("Response missing rqid: protocolId={}, body={}", 
                        protocolId, jsonBody.length() > 100 ? jsonBody.substring(0, 100) + "..." : jsonBody);
                }
            } catch (Exception parseEx) {
                logger.error("Failed to parse response JSON: protocolId={}, error={}, body={}", 
                    protocolId, parseEx.getMessage(), jsonBody.length() > 100 ? jsonBody.substring(0, 100) + "..." : jsonBody);
            }
            
        } catch (Exception e) {
            logger.error("Failed to handle response: {}", e.getMessage(), e);
        }
    }
    
    public void disconnect() {
        try {
            connected = false;
            stopHeartbeat(); // 停止心跳
            if (channel != null) {
                channel.close();
            }
            if (group != null) {
                group.shutdownGracefully();
            }
            heartbeatExecutor.shutdown(); // 关闭心跳线程池
        } catch (Exception e) {
            logger.error("Error during disconnect: {}", e.getMessage());
        }
    }
    
    /**
     * 启动心跳保活任务
     */
    private void startHeartbeat() {
        if (heartbeatTask != null) {
            heartbeatTask.cancel(false);
        }
        
        heartbeatTask = heartbeatExecutor.scheduleAtFixedRate(() -> {
            try {
                sendHeartbeat();
            } catch (Exception e) {
                logger.warn("心跳发送失败: {}", e.getMessage());
            }
        }, heartbeatInterval, heartbeatInterval, TimeUnit.SECONDS);
        
        logger.debug("心跳保活已启动，间隔: {}秒", heartbeatInterval);
    }
    
    /**
     * 停止心跳保活任务
     */
    private void stopHeartbeat() {
        if (heartbeatTask != null) {
            heartbeatTask.cancel(false);
            heartbeatTask = null;
            logger.debug("心跳保活已停止");
        }
    }
    
    /**
     * 发送心跳消息
     */
    private void sendHeartbeat() {
        if (!isConnected()) {
            return;
        }
        
        try {
            // 使用标准sendMessage方法，自动包装UUID和data结构
            Map<String, Object> heartbeatData = Map.of("timestamp", System.currentTimeMillis());
            
            // 异步发送心跳，不等待响应避免阻塞心跳线程
            sendMessage((short) 9001, heartbeatData, 2, TimeUnit.SECONDS)
                .whenComplete((response, throwable) -> {
                    if (throwable != null) {
                        logger.debug("心跳响应异常: {}", throwable.getMessage());
                    } else {
                        logger.trace("心跳响应正常");
                    }
                });
            
        } catch (Exception e) {
            logger.warn("心跳消息发送失败: {}", e.getMessage());
        }
    }
}
